package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

/**
 * Created by Smexy on 2023/4/10
 *
 *  是有状态的运算，状态需要一直存储在当前TaskManager的内存中。
 *
 *      可以设置ttl(time to live 存活时间)时间，那些不经常访问的key，所存储的状态就可以被清除，节省内存。
 *
 *      类似redis的内存回收策略。
 *
 *      前提，状态是存储在内存中，内存空间不够，采取过期策略。
 *      状态不在内存中存储，可以斟酌，是否有必要。
 *
 *
 */
public class Demo10_StateTtl
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建状态的存储策略
        StateTtlConfig stateTtlConfig = StateTtlConfig
            //就是processingTime中的时间，不参考水印
            .newBuilder(Time.seconds(10))
            //当前状态已经过期了，在查询时，就不返回
            .neverReturnExpired()
            /*
                 如何判断当前状态已经过期?
                    如果一个状态，应用了StateTtlConfig策略。
                    那么状态在存储时，会额外存储一个访问时间字段。

                        之前，没有设置ttl:
                            s1 ---->[1,2,3]
                        现在，设置了ttl:
                             s1 ---->[1,2,3]，accessTime= 2023-04-10 15：00：00


                   访问时间的更新策略，取决于StateTtlConfig.UpdateType.xxx
                        OnReadAndWrite: 读状态或写状态时，都会去更新accessTime。

                   当读状态或写状态时，会获取到状态的上次 accessTime，和当前的物理时钟，相减，之后差值超过  Time ttl，此时状态就是过期了，根据策略，不返回，返回空。

                   如果这是一个僵尸状态： 创建时使用了一次，后续再无使用，如何判断它是否过期?
                        flink后台会维护一个线程，定时判断，哪些状态已经过期。过期后，会删除状态，释放内存空间。
             */
            .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
            .build();

        env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction())
            .keyBy(WaterSensor::getId)
            .process(new KeyedProcessFunction<String, WaterSensor, String>()
            {
                //用来存储当前前三水位的 集合
                private ListState<Integer> top3;

                // keyed state，都是在当前Task初始化时，从备份中获取状态。
                @Override
                public void open(Configuration parameters) throws Exception {
                    ListStateDescriptor<Integer> stateDescriptor = new ListStateDescriptor<>("top3", Integer.class);

                    //添加过期策略
                    stateDescriptor.enableTimeToLive(stateTtlConfig);

                    this.top3 = getRuntimeContext().getListState(stateDescriptor);
                }

                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                    //把当前元素添加到集合中
                    top3.add(value.getVc());
                    //在集合中进行排序和对比，取前3
                    List<Integer> top3Vc = StreamSupport.stream(this.top3.get().spliterator(), true)
                                                        .sorted(Comparator.reverseOrder())
                                                        .limit(3)
                                                        .collect(Collectors.toList());

                    //只存前三名
                    top3.update(top3Vc);

                    out.collect(ctx.getCurrentKey() + "的前3:" +top3Vc.toString());

                }
            })
            .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
